package com.poetic.network.flow.function;

import com.poetic.network.flow.domain.ItemViewCount;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * <pre>
 *  用于输出窗口的结果
 * Created by lianghuikun on 2020-09-14.
 * </pre>
 *
 * @author lianghuikun
 */
public class WindowResultFunction implements WindowFunction<Long, ItemViewCount, Long, TimeWindow> {

    @Override
    public void apply(Long key, TimeWindow window, Iterable<Long> input, Collector<ItemViewCount> out) throws Exception {
        Long count = input.iterator().next();
        ItemViewCount itemViewCount = ItemViewCount.builder()
                .itemId(key)
                .count(count)
                .windowEnd(window.getEnd())
                .build();
        out.collect(itemViewCount);
    }
}
